package io.kiki.sba.registry.server.clientInterface.remoting.handler;


import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.server.clientInterface.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.clientInterface.registry.Registry;
import io.kiki.sba.registry.server.shared.remoting.ListenServerChannelHandler;
import io.kiki.sba.registry.util.AtomicSet;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Set;


public class ClientNodeConnectionHandler extends ListenServerChannelHandler implements ApplicationListener<ContextRefreshedEvent> {
    private final Logger logger = LoggerFactory.getLogger(ClientNodeConnectionHandler.class);
    private final AtomicSet<ClientToClientInterfaceServerNodeId> pendingClientOff = new AtomicSet<>();
    private final ClientOffWorker clientOffWorker = new ClientOffWorker();
    @Autowired
    Registry sessionRegistry;
    @Autowired
    ExecutorManager executorManager;
    private volatile boolean stopped = false;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        start();
    }

    public void start() {
        ConcurrentUtils.createDaemonThread("ClientOff-Worker", clientOffWorker).start();
    }

    @Override
    public void disconnected(Channel channel) {
        if (stopped) {
            return;
        }
        super.disconnected(channel);
        fireCancelClient(channel);
    }

    public void stop() {
        this.stopped = true;
    }

    @Override
    protected Node.NodeType getConnectNodeType() {
        return Node.NodeType.client;
    }

    void fireCancelClient(Channel channel) {
        pendingClientOff.add(ClientToClientInterfaceServerNodeId.of(channel.getRemoteAddress(), channel.getLocalAddress()));
        clientOffWorker.wakeup();
    }

    private class ClientOffWorker extends WakeUpLoopExecuteTask {
        @Override
        public void _execute_() {
            Set<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds = pendingClientOff.getAndReset();
            if (!CollectionUtils.isEmpty(clientToClientInterfaceServerNodeIds)) {
                long start = System.currentTimeMillis();
                sessionRegistry.clean(new ArrayList<>(clientToClientInterfaceServerNodeIds));
                long span = System.currentTimeMillis() - start;
                logger.info("disconnect size={},span={}", clientToClientInterfaceServerNodeIds.size(), span);
            }
        }

        @Override
        public int getWaitingMillis() {
            return 5000;
        }
    }
}
